#include "MediaSourceCacheCenter.h"
#include "Util/util.h"
#include "Rtp/RtpSelector.h"

#ifdef ENABLE_CACHE
#if defined(NDEBUG)
#define MAX_MEM_OCCUPY_RATIO 1.2    //内存占用超过80%就禁用
#define MIN_MEM_OCCUPY_RATIO 5      //内存占用低于20%就重新解禁缓存功能
#else
#define MAX_MEM_OCCUPY_RATIO 50     //测试版本内存占用超过2%就禁用
#define MIN_MEM_OCCUPY_RATIO 80      //测试版本内存占用低于1.25%就禁用
#endif
namespace mediakit {
    INSTANCE_IMP(MediaSourceCacheCenter);
    
    static inline int get_memory_by_pid(pid_t pid) {
        FILE* fd;
        char line[1024] = {0};
        char virtual_filename[32] = {0};
        char vmrss_name[32] = {0};
        int vmrss_num = 0;
        sprintf(virtual_filename, "/proc/%d/status", pid);
        fd = fopen(virtual_filename, "r");
        if(fd == NULL) {
            cout << "open " << virtual_filename << " failed" << endl;
            abort();
        }
        // VmRSS line is uncertain
        for(int i = 0 ; ; i++) {
            if(fgets(line,sizeof(line),fd) == NULL) {
                break;
            }
            if(strstr(line, "VmRSS:") != NULL) {
                sscanf(line, "%s %d", vmrss_name, &vmrss_num);
                break;
            }
        }
        fclose(fd);
        return vmrss_num;
    }
    
    static inline int get_machine_memory() {
        const char* virtual_filename = "/proc/meminfo";
        FILE* fd;
        char line[1024] = {0};
        fd = fopen(virtual_filename, "r");
        if(fd == NULL) {
            cout << "open " << virtual_filename << " failed" << endl;
            abort();
        }
        char vmrss_name[32];
        int machine_memory;
        // MemTotal line is uncertain
        for(int i = 0 ; ; i++) {
            if(fgets(line,sizeof(line),fd) == NULL) {
                break;
            }
            if(strstr(line, "MemTotal:") != NULL) {
                sscanf(line, "%s %d", vmrss_name, &machine_memory);
                break;
            }
        }
        fclose(fd);
        return machine_memory;
    }

    DeviceRecordCacheHelper::DeviceRecordCacheHelper(const privateproto::DeviceChannel& dev_chan, bool autoscribe) {
        m_dev_chan = dev_chan;
        m_cache_reord = std::make_shared<DeviceRecordCache>(m_dev_chan,autoscribe);
    }
    
    DeviceRecordCacheHelper::~DeviceRecordCacheHelper() {
        
    }
    
    DeviceRecordCache::Ptr &DeviceRecordCacheHelper::getCacher() {
        return m_cache_reord;
    }
    int64_t DeviceRecordCacheHelper::PurgeMemory(const int machine_memory,const int process_memory) {
        return m_cache_reord->PurgeMemory(machine_memory,process_memory);
    }
    int64_t DeviceRecordCacheHelper::DeviceCacheMemorySize(){
        return m_cache_reord->DeviceCacheMemorySize();
    }
    void DeviceRecordCacheHelper::CleanCache(){
        m_cache_reord->CleanCache();
    }
    Value DeviceRecordCacheHelper::DumpAsJson(){
        Value root;
        Value media;
        media["deviceid"] = m_dev_chan._deviceid;
        media["channelid"] = m_dev_chan._channelid;
        media["vhost"] = m_dev_chan._vhost;
        root["media"] = media;
        root["cache"] = m_cache_reord->DumpAsJson();
        return root;
    }
    void DeviceRecordCacheHelper::DumpAsJsonInfo(const function<void(const Json::Value& info)> &cb){
        Value root;
        Value media;
        media["deviceid"] = m_dev_chan._deviceid;
        media["channelid"] = m_dev_chan._channelid;
        media["vhost"] = m_dev_chan._vhost;
        root["media"] = media;
        m_cache_reord->DumpAsJsonInfo([cb,root](const Json::Value& info) mutable{
            root["cache"] = info;
            cb(root);
        });
    }
    const int MediaSourceCacheCenter::m_machine_memory = get_machine_memory();;
    MediaSourceCacheCenter::MediaSourceCacheCenter() {
        m_poller = CacheThreadPool::Instance().getPoller();
            m_check_timer = std::make_shared<Timer>(10.0f,[&](){
            TimeTicker();    
            //TraceL << "机器内存:" << m_machine_memory << ",进程内存:" << get_memory_by_pid(getpid());  
            lock_guard<decltype(m_mtx_map)> lck(m_mtx_map);
            int total_device_cached_mem = 0;
            for(auto it = m_map_cache_reord.begin();it != m_map_cache_reord.end();it++) {
                total_device_cached_mem += (int)(it->second->DeviceCacheMemorySize()/1024);
            }
            int process_memory = get_memory_by_pid(getpid());
            #if !defined(NDEBUG)
            InfoL << "机器内存:" << m_machine_memory << ",缓存内存:" << total_device_cached_mem << " 进程内存:" << process_memory;
            #endif
            for(auto it = m_map_cache_reord.begin();it != m_map_cache_reord.end();) {
                int process_memory = get_memory_by_pid(getpid());
                int64_t erased_memory = it->second->PurgeMemory(m_machine_memory,process_memory);
                if(erased_memory < 0){
                    m_map_cache_reord.erase(it++);
                }else{
                    if(m_machine_memory < total_device_cached_mem*MAX_MEM_OCCUPY_RATIO){
                        m_bmemreachLimit.store(true);
                        std::string stream_id = it->first.GenStreamId();
                        auto rtp_process = RtpSelector::Instance().getProcess(stream_id,0,0, false);
                        if (rtp_process) {
                            rtp_process->setStopRtpCache(true);
                            #if !defined(NDEBUG)
                            InfoL << "机器内存:" << m_machine_memory << ",缓存内存:" << total_device_cached_mem << " 关闭缓存:" << stream_id;
                            #endif
                        }/*else{
                            ErrorL << "机器内存:" << m_machine_memory << ",缓存内存:" << total_device_cached_mem << " 关闭缓存查找不到流:" << stream_id;
                        }*/
                    }else {
                        if(m_machine_memory > total_device_cached_mem*MIN_MEM_OCCUPY_RATIO){//再次低于下水位的时候就置为可用
                            m_bmemreachLimit.store(false);
                        }
                        if(m_machine_memory < process_memory*MAX_MEM_OCCUPY_RATIO){
                            std::string stream_id = it->first.GenStreamId();
                            auto rtp_process = RtpSelector::Instance().getProcess(stream_id,0,0, false);
                            if (rtp_process) {
                                rtp_process->setStopRtpCache(true);
                                #if !defined(NDEBUG)
                                InfoL << "机器内存:" << m_machine_memory << ",进程内存:" << get_memory_by_pid(getpid()) << " 关闭缓存:" << stream_id;
                                #endif
                            }/*else{
                                ErrorL << "机器内存:" << m_machine_memory << ",进程内存:" << get_memory_by_pid(getpid()) << " 关闭缓存查找不到流:" << stream_id;
                            }*/
                        }
                    }
                    it++;
                }
            }
            return true;
            }, m_poller);
    }
    bool MediaSourceCacheCenter::CheckMemReachLimit() const {
        return m_bmemreachLimit.load();
    }
    MediaSourceCacheCenter::~MediaSourceCacheCenter() {
        m_check_timer = nullptr;
    }
    DeviceRecordCache::Ptr MediaSourceCacheCenter::GetRecordCacher(const privateproto::DeviceChannel& dev_chan,bool bcreate, bool autoscribe) {
        lock_guard<decltype(m_mtx_map)> lck(m_mtx_map);
        auto it = m_map_cache_reord.find(dev_chan);
        if (it == m_map_cache_reord.end()) {
            if(bcreate) {
                m_map_cache_reord[dev_chan] = std::make_shared<DeviceRecordCacheHelper>(dev_chan,autoscribe);
                TraceL << "创建缓存对象: " << dev_chan._deviceid << "|" << dev_chan._channelid << "|" << dev_chan._ssrc;
            }else {
                return nullptr;
            }
        }
        DeviceRecordCacheHelper::Ptr &ref = m_map_cache_reord[dev_chan];
        return ref->getCacher();
    }
    
    void MediaSourceCacheCenter::DumpAsJsonInfo(const std::string & channel_id,const function<void(const Json::Value& info)> &cb) {
        lock_guard<decltype(m_mtx_map)> lck(m_mtx_map);
        if(m_map_cache_reord.size() == 0) {
            Json::Value root(Json::arrayValue);
            cb(root);
            return;
        }
        auto it = m_map_cache_reord.begin();
        if(channel_id.size() > 5) {
            for(;it != m_map_cache_reord.end();it++) {
                if(it->first._channelid.compare(channel_id) == 0) {
                    Json::Value root(Json::arrayValue);
                    Value item;
                    item["deviceid"] = it->first._deviceid;
                    item["channelid"] = it->first._channelid;
                    it->second->DumpAsJsonInfo([cb,item,root](const Json::Value& info) mutable{
                        item["cache"] = info;
                        root.append(item);
                        cb(root);
                    });
                    break;
                }
            }
            if(it == m_map_cache_reord.end()) {
                Json::Value root(Json::arrayValue);
                Value item;
                item["deviceid"] = "device not found";
                item["channelid"] = channel_id;
                item["error"] = "cache not found";
                root.append(item);
                cb(root);
            }
        }else{
            Json::Value root(Json::arrayValue);
            size_t record_count = {0};
            size_t record_size = m_map_cache_reord.size();
            for(;it != m_map_cache_reord.end();it++) {
                Value item;
                item["deviceid"] = it->first._deviceid;
                item["channelid"] = it->first._channelid;
                root.append(item);
            }
            cb(root);
        }
    }
} /* namespace toolkit */

#endif
